/**
 * Copyright © 1998-2020, Glodon Inc. All Rights Reserved.
 */
package reactor2;

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 此处填写类简介
 * <p>
 * 此处填写类说明
 * </p>
 * 
 * @author dujl
 * @since jdk1.6 2020年8月6日
 */

public class App {
	public static void main(String[] args) throws InterruptedException {
		// core 20 max 50 5s 回收
//		ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
//		Demultiplexer<Event> exe = new Demultiplexer<Event>();
//		EventDispatcher eventDispatcher = new EventDispatcher();
		EventContext context =  new EventContext();
		initRegister(context);
		//initRegister(eventDispatcher);
		// 多路复用
		//Selectlor selector = new Selectlor(exe, eventDispatcher, executor);
		//启动，模拟多个并发输入点
		start(context);
	}

	/***
	 * TODO 注册事件
	 */
	public static void initRegister(EventContext context) {
		// EventType type, IEventHandler handler
		EventDispatcher pather = context.getEventDispatcher();
		pather.registerHandler(EventType.LOG, new LogEventHandler());
		pather.registerHandler(EventType.KAFKA, new KafkaEventHandler());
	}

	/***
	 * TODO
	 * 
	 * @throws InterruptedException
	 */
/*	public static void start(ThreadPoolExecutor executor, final Demultiplexer dem) throws InterruptedException {
		for (int i = 1; i <= 100; i++) {
			final Event logEvent = new Event(EventType.LOG, null);
			final Event KafkaEvent = new Event(EventType.KAFKA, null);
			Thread command = new Thread(new Runnable() {
				@Override
				public void run() {
					// TODO Auto-generated method stub
					dem.addMsg(logEvent);
					dem.addMsg(KafkaEvent);
				}
			});
			executor.execute(command);
		}
		
		for(;;){
			Thread.sleep(1000);
			 Event logEvent = new Event(EventType.LOG, null);
			 Event KafkaEvent = new Event(EventType.KAFKA, null);
			dem.addMsg(logEvent);
			dem.addMsg(KafkaEvent);
		}
	}*/
	
	public static void start(EventContext context) throws InterruptedException {
		Demultiplexer dem = context.getDemultiplexer();
		for (int i = 1; i <= 100; i++) {
			final Event logEvent = new Event(EventType.LOG, null);
			final Event KafkaEvent = new Event(EventType.KAFKA, null);
			Thread command = new Thread(new Runnable() {
				@Override
				public void run() {
					// TODO Auto-generated method stub
					dem.addMsg(logEvent);
					dem.addMsg(KafkaEvent);
				}
			});
			context.getExecutor().execute(command);
		}
		
		for(;;){
			Thread.sleep(1000);
			 Event logEvent = new Event(EventType.LOG, null);
			 Event KafkaEvent = new Event(EventType.KAFKA, null);
			dem.addMsg(logEvent);
			dem.addMsg(KafkaEvent);
		}
	}
}
